package studydemo.service.listener;


import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.stereotype.Service;

import java.util.HashMap;
import java.util.Map;

/**
 * 消息确认机制
 * https://blog.csdn.net/K_kzj_K/article/details/106642250
 *
 * 监听多个队列的手动确认
 * 消费者接收到消息的消息确认机制。
 * 和生产者的消息确认机制不同，因为消息接收本来就是在监听消息，符合条件的消息就会消费下来。
 * 所以，消息接收的确认机制主要存在三种模式：
 *
 * ①自动确认， 这也是默认的消息确认情况。  AcknowledgeMode.NONE
 * RabbitMQ成功将消息发出（即将消息成功写入TCP Socket）中立即认为本次投递已经被正确处理，不管消费者端是否成功处理本次投递。
 * 所以这种情况如果消费端消费逻辑抛出异常，也就是消费端没有处理成功这条消息，那么就相当于丢失了消息。
 * 一般这种情况我们都是使用try catch捕捉异常后，打印日志用于追踪数据，这样找出对应数据再做后续处理。
 *
 * ② 根据情况确认， 这个不做介绍
 * ③ 手动确认 ， 这个比较关键，也是我们配置接收消息确认机制时，多数选择的模式。
 * 消费者收到消息后，手动调用basic.ack/basic.nack/basic.reject后，RabbitMQ收到这些消息后，才认为本次投递成功。
 * basic.ack用于肯定确认
 * basic.nack用于否定确认（注意：这是AMQP 0-9-1的RabbitMQ扩展）
 * basic.reject用于否定确认，但与basic.nack相比有一个限制:一次只能拒绝单条消息
 *
 * 消费者端以上的3个方法都表示消息已经被正确投递，但是basic.ack表示消息已经被正确处理。
 * 而basic.nack,basic.reject表示没有被正确处理：
 *
 * 着重讲下reject，因为有时候一些场景是需要重新入列的。
 *
 * channel.basicReject(deliveryTag, true);  拒绝消费当前消息，如果第二参数传入true，就是将数据重新丢回队列里，那么下次还会消费这消息。设置false，就是告诉服务器，我已经知道这条消息数据了，因为一些原因拒绝它，而且服务器也把这个消息丢掉就行。 下次不想再消费这条消息了。
 *
 * 使用拒绝后重新入列这个确认模式要谨慎，因为一般都是出现异常的时候，catch异常再拒绝入列，选择是否重入列。
 *
 * 但是如果使用不当会导致一些每次都被你重入列的消息一直消费-入列-消费-入列这样循环，会导致消息积压。
 *
 *
 *
 * 顺便也简单讲讲 nack，这个也是相当于设置不消费某条消息。
 *
 * channel.basicNack(deliveryTag, false, true);
 * 第一个参数依然是当前消息到的数据的唯一id;
 * 第二个参数是指是否针对多条消息；如果是true，也就是说一次性针对当前通道的消息的tagID小于当前这条消息的，都拒绝确认。
 * 第三个参数是指是否重新入列，也就是指不确认的消息是否重新丢回到队列里面去。
 *
 * 同样使用不确认后重新入列这个确认模式要谨慎，因为这里也可能因为考虑不周出现消息一直被重新丢回去的情况，导致积压。
 *
 * @author xn
 * @date 2022/11/21 20:27
 */
@Service
public class MyAckReceiverMethod2Service implements ChannelAwareMessageListener {

    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        try {
            //因为传递消息的时候用的map传递,所以将Map从Message内取出需要做些处理
            String msg = message.toString();

            //可以点进Message里面看源码,单引号直接的数据就是我们的map消息数据
            String[] msgArray = msg.split("'");

            Map<String, String> msgMap = mapStringToMap(msgArray[1].trim());
            String messageId = msgMap.get("messageId");
            String messageData = msgMap.get("messageData");
            String createTime = msgMap.get("createTime");

            if ("TestDirectQueue".equals(message.getMessageProperties().getConsumerQueue())) {
                System.out.println();
                System.out.println("监听队列，手动确认");
                System.out.println("消费的消息来自的队列名为：" + message.getMessageProperties().getConsumerQueue());
                System.out.println("消息成功消费到  messageId:" + messageId + "  messageData:" + messageData + "  createTime:" + createTime);
                System.out.println("执行TestDirectQueue中的消息的业务处理流程......");
                System.out.println();

            }

            if ("fanout.A".equals(message.getMessageProperties().getConsumerQueue())) {
                System.out.println();
                System.out.println("监听队列，手动确认");
                System.out.println("消费的消息来自的队列名为：" + message.getMessageProperties().getConsumerQueue());
                System.out.println("消息成功消费到  messageId:" + messageId + "  messageData:" + messageData + "  createTime:" + createTime);
                System.out.println("执行fanout.A中的消息的业务处理流程......");
                System.out.println();

            }

            //这个设置消息不会再放回队列中
            //channel.basicAck(deliveryTag, true);
            //为true会重新放回队列
            channel.basicReject(deliveryTag, true);
        } catch (Exception e) {
            channel.basicReject(deliveryTag, false);
            e.printStackTrace();
        }
    }

    //{key=value,key=value,key=value} 格式转换成map
    private Map<String, String> mapStringToMap(String str) {

        str = str.substring(1, str.length() - 1);

        String[] strs = str.split(",");

        Map<String, String> map = new HashMap<String, String>();
        //这种写解决了，send消息的时候，消息内容里面不能有,号的问题
        for (int i = 0; i < 3; i++) {
            String key = strs[i].split("=")[0].trim();
            String value = strs[i].split("=")[1];
            if ( i == 2 && strs.length > 3) {
                for (int j = 3; j < strs.length; j++) {
                    System.out.println(strs[j]);
                    value = strs[i].concat(",").concat(strs[j]);
                }
            }
            map.put(key,value);
        }

        return map;
    }
}
